package il.technion.ewolf.beehive.handlers;

import il.technion.ewolf.beehive.cache.DataCollector;
import il.technion.ewolf.beehive.cache.Storage;
import il.technion.ewolf.kbr.Node;
import il.technion.ewolf.kbr.openkad.bucket.KBuckets;
import il.technion.ewolf.kbr.openkad.cache.KadCache;
import il.technion.ewolf.kbr.openkad.handlers.AbstractHandler;
import il.technion.ewolf.kbr.openkad.msg.FindValueRequest;
import il.technion.ewolf.kbr.openkad.msg.FindValueResponse;
import il.technion.ewolf.kbr.openkad.msg.KadMessage;
import il.technion.ewolf.kbr.openkad.net.Communicator;
import il.technion.ewolf.kbr.openkad.net.MessageDispatcher;
import il.technion.ewolf.kbr.openkad.net.filter.MessageFilter;
import il.technion.ewolf.kbr.openkad.net.filter.TypeMessageFilter;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.name.Named;

/**
 * Handle find value requests by giving the known closest nodes to the requested key
 * from the KBuckets data structure
 * @author Ligal ligal2@gmail.com
 * @author Hen senderhen@gmail.com
 */
public class BeeHiveFindValueHandler extends AbstractHandler implements FindValueHandler {
	private final Communicator kadServer;
	private final Node localNode;
	private final Storage localStorage;
	private final KadCache cache;
	private final KBuckets kBuckets;
	private final int kBucketSize;
	private final DataCollector statistics;
	
	private final AtomicInteger nrFindValueHits;
	private final AtomicInteger nrFindValueMiss;
	
	
	@Inject
	BeeHiveFindValueHandler(
			Provider<MessageDispatcher<Void>> msgDispatcherProvider,
			Communicator kadServer,
			@Named("openkad.local.node") Node localNode,
			Storage storage,
			KadCache cache,
			DataCollector p_statistics,
			KBuckets kBuckets,
			@Named("openkad.bucket.kbuckets.maxsize") int kBucketSize,
			
			@Named("openkad.testing.nrFindValueHits") AtomicInteger nrFindValueHits,
			@Named("openkad.testing.nrFindValueMiss") AtomicInteger nrFindValueMiss) {
		
		super(msgDispatcherProvider);
		this.kadServer = kadServer;
		this.localNode = localNode;
		this.localStorage = storage;
		this.cache = cache;
		this.kBuckets = kBuckets;
		this.kBucketSize = kBucketSize;
		this.statistics = p_statistics;
		
		this.nrFindValueHits = nrFindValueHits;
		this.nrFindValueMiss = nrFindValueMiss;
	}

	@Override
	public void completed(KadMessage msg, Void attachment) {
		
		FindValueRequest findValueRequest = ((FindValueRequest)msg);
		FindValueResponse findValueResponse = findValueRequest
				.generateResponse(localNode)
				.setStoredResult(false)
				.setContainingResult(false);
		
		Serializable storedResults = null;
		//add to the statistic if this is not replication
		if (!findValueRequest.getFlag())
			statistics.increse(findValueRequest.getKey());
		
		//first search in storage
		storedResults = localStorage.search(findValueRequest.getKey());
					
		if (storedResults == null) {
			List<Node> cachedResults = null;
			
			//search in cache
			cachedResults = cache.search(findValueRequest.getKey());
			if (cachedResults == null) {
				nrFindValueMiss.incrementAndGet();
				findValueResponse.setNodes(kBuckets.getClosestNodesByKey(
							findValueRequest.getKey(), kBucketSize));
			} else {
				nrFindValueHits.incrementAndGet();
				findValueResponse
					.setNodes(new ArrayList<Node>(cachedResults))
					.setStoredResult(true);						
			}
			

		} else {
			nrFindValueHits.incrementAndGet();
			findValueResponse
				.setValue(storedResults)
				.setStoredResult(true)
				.setContainingResult(true);
				
		}
		
		
		try {
			kadServer.send(msg.getSrc(), findValueResponse);
		} catch (IOException e) {
			// could not send back a response
			// nothing to do
			e.printStackTrace();
		}
	}

	@Override
	public void failed(Throwable exc, Void attachment) {
		// should never b here
	}

	@Override
	protected Collection<MessageFilter> getFilters() {
		// only accept FindNodeRequests messages
		return Arrays.asList(new MessageFilter[] {
				new TypeMessageFilter(FindValueRequest.class)
		});
	}
}
